package com.atguigu.dataStreamApi.transformation;

/*
* 分流
*
*   1.简单实现，使用filter算子实现
*
*   2.使用侧输出流实现
*
*
*
* */

import com.atguigu.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import util.SourceUtil;

public class Flink07_SplitStream {

   public static void main(String[] args) {
       StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
       env.setParallelism(1);
       //需求：将Tom分到一条流，Jerry分到一条流，其他的分到一条流

       //1.对接数据源，利用数据生成器模拟生成数据
        DataStreamSource<Event> ds
                = env.fromSource(SourceUtil.getSource(), WatermarkStrategy.noWatermarks(), "dstaGenSource");

       // 定义输出标签，侧输出流的数据类型为三元组(user, url, timestamp)
       OutputTag<Event> tomTag = new OutputTag<Event>("tomTag",Types.POJO(Event.class));
       OutputTag<Event> jerryTag = new OutputTag<Event>("jerryTom",Types.POJO(Event.class));
       OutputTag<Event> otherTag = new OutputTag<Event>("otherTag",Types.POJO(Event.class));

       //数据转换处理
       SingleOutputStreamOperator<Event> processDs = ds.process(
               new ProcessFunction<Event, Event>() {
                   @Override
                   public void processElement(Event value, ProcessFunction<Event, Event>.Context ctx, Collector<Event> out) throws Exception {
                       //这种感觉怎么像是还在一条河流里，只是贴上了不同的标签呢
                       if ("Tom".equals(value.getUser())){
                           //分到Tom流
                           ctx.output(tomTag,value); //之前已经定义过标签了

                       } else if ("Jerry".equals(value.getUser())) {
                           //分到jerry流
                           ctx.output(jerryTag,value);
                       }else {
                           //分到其它流,一共三种情况，最后一个保存在主流里
                           out.collect(value);
                       }
                   }
               }
       );
       //捕获测流
       SideOutputDataStream<Event> tomDs  = processDs.getSideOutput(tomTag);
       SideOutputDataStream<Event> jerryDs = processDs.getSideOutput(jerryTag);

       //打印流
       tomDs.print("Tom");
       jerryDs.print("Jerry");
       processDs.print("MAIN");


       try {
               env.execute();
           } catch (Exception e) {
               throw new RuntimeException(e);
           }
       }

}
